use crate::scheduler::prelude::SystemScheduled;
use crate::scheduler::system_scheduled::{ActiveModel, Model};
use crate::scheduler::{JOB_FUNCTIONS, SCHEDULER, system_scheduled};
use crate::util::next_id;
use crate::{ApiError, ApiResult, ToError, db};
use chrono::Local;
use cron::Schedule;
use sea_orm::prelude::DateTime;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use std::sync::{Arc, LazyLock};
use std::time::Duration;
use tokio::sync::Mutex;
use tokio_cron_scheduler::{Job, JobScheduler};

static LOCK: LazyLock<Arc<Mutex<()>>> = LazyLock::new(|| Arc::new(Mutex::new(())));

/// 定时任务相关参数
pub struct Scheduled {
    #[doc = "业务id"]
    pub business_id: i64,
    #[doc = "业务类型"]
    pub business_type: Option<i8>,
    #[doc = "cron表达式"]
    pub cron: Option<String>,
    #[doc = "定时任务执行的函数的唯一标识"]
    pub task: String,
    #[doc = "开始执行的时间"]
    pub start_time: Option<DateTime>,
}

/// 保存定时任务
pub async fn save_scheduled(scheduled: Scheduled) -> ApiResult<()> {
    let task = scheduled.task;

    let now = Local::now().naive_local();
    let business_type = scheduled.business_type.unwrap_or(-1);
    let model = get_scheduled(business_type, scheduled.business_id).await?;
    let model = match model {
        Some(model) => {
            let mut am: ActiveModel = model.into();
            am.cron = Set(scheduled.cron);
            am.task = Set(task);
            am.start_time = Set(scheduled.start_time);
            am.update(db::get())
                .await
                .to_err(|| ApiError::Biz("更新定时任务失败".into()))?
        }
        None => {
            let am = ActiveModel {
                id: Set(next_id()),
                business_id: Set(scheduled.business_id),
                business_type: Set(business_type),
                cron: Set(scheduled.cron),
                task: Set(task),
                start_time: Set(scheduled.start_time),
                del_flag: Set(false),
                create_time: Set(now),
                update_time: Set(now),
            };
            am.insert(db::get())
                .await
                .to_err(|| ApiError::Biz("保存定时任务失败".into()))?
        }
    };
    start(model).await
}

/// 删除定时任务
pub async fn delete_scheduled(business_id: i64, business_type: Option<i8>) -> ApiResult<()> {
    let business_type = business_type.unwrap_or(-1);
    let model = SystemScheduled::find()
        .filter(system_scheduled::Column::BusinessType.eq(business_type))
        .filter(system_scheduled::Column::BusinessId.eq(business_id))
        .filter(system_scheduled::Column::DelFlag.eq(false))
        .one(db::get())
        .await
        .to_err(|| ApiError::Biz("查询任务错误".into()))?
        .to_err(|| ApiError::Biz("任务不存在".into()))?;
    let _lock = LOCK.lock().await;
    let id = model.id;
    let mut am: ActiveModel = model.into();
    am.del_flag = Set(true);
    am.update(db::get())
        .await
        .to_err(|| ApiError::Biz("删除任务失败".into()))?;
    let scheduler = SCHEDULER.remove(&id);

    // 取消定时任务
    if let Some(mut scheduler) = scheduler {
        scheduler
            .1
            .shutdown()
            .await
            .to_err(|| ApiError::Biz("取消定时任务错误".into()))?;
        SCHEDULER.remove(&id);
    }

    Ok(())
}

/// 开始定时任务
async fn start(model: Model) -> ApiResult<()> {
    let _lock = LOCK.lock().await;

    // 如果之前有定时任务，则取消
    let scheduler = SCHEDULER.get(&model.id);
    if let Some(scheduler) = scheduler {
        let mut scheduler = scheduler.clone();
        scheduler
            .shutdown()
            .await
            .to_err(|| ApiError::Biz("取消定时任务错误".into()))?;
    }
    let sched = JobScheduler::new()
        .await
        .to_err(|| ApiError::Biz("创建调度器失败".into()))?;
    let task = model.task;
    let job_function = *JOB_FUNCTIONS
        .get(&task)
        .to_err(|| ApiError::Biz("函数不存在".into()))?;

    let cron = model.cron;
    let start_time = model.start_time;
    let business_id = model.business_id;
    let job = if let Some(cron) = cron {
        Schedule::try_from(cron.as_str()).to_err(|| ApiError::Biz("cron表达式错误".into()))?;
        Job::new_async(cron, move |_, _| {
            Box::pin(async move {
                job_function(business_id).await;
            })
        })
        .to_err(|| ApiError::Biz("创建任务错误".into()))?
    } else {
        let time = start_time.unwrap().and_local_timezone(Local).unwrap();
        let delta = time - Local::now();
        let mut sec = delta.num_seconds();
        if sec < 0 {
            sec = 0;
        }
        Job::new_one_shot_async(Duration::from_secs(sec as u64), move |_, _| {
            Box::pin(async move {
                job_function(business_id).await;
            })
        })
        .to_err(|| ApiError::Biz("创建任务错误".into()))?
    };

    sched
        .add(job)
        .await
        .to_err(|| ApiError::Biz("添加任务错误".into()))?;
    sched
        .start()
        .await
        .to_err(|| ApiError::Biz("开始定时任务".into()))?;

    SCHEDULER.insert(model.id, sched);
    Ok(())
}

/// 项目启动时初始化定时任务
pub async fn init_scheduler() {
    let models = SystemScheduled::find()
        .filter(system_scheduled::Column::DelFlag.eq(false))
        .all(db::get())
        .await
        .expect("查询任务失败");

    for model in models {
        start(model).await.expect("启动任务失败");
    }
}

async fn get_scheduled(business_type: i8, business_id: i64) -> ApiResult<Option<Model>> {
    SystemScheduled::find()
        .filter(system_scheduled::Column::BusinessType.eq(business_type))
        .filter(system_scheduled::Column::BusinessId.eq(business_id))
        .filter(system_scheduled::Column::DelFlag.eq(false))
        .one(db::get())
        .await
        .to_err(|| ApiError::Biz("查询任务错误".into()))
}
